package com.dabai.iot.mqtt.server;

import com.dabai.common.core.utils.StringUtils;
import com.dabai.iot.mqtt.event.DeviceConnectStatusListener;
import com.dabai.iot.mqtt.event.DeviceMessageListener;
import com.dabai.iot.mqtt.event.ServerConnectStatusListener;
import com.dabai.iot.mqtt.event.ServerMessageListener;
import com.dabai.iot.mqtt.model.ServerClient;
import com.dabai.iot.mqtt.util.MqttSessionManager;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.vertx.mqtt.MqttAuth;
import io.vertx.mqtt.MqttEndpoint;
import io.vertx.mqtt.MqttTopicSubscription;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class MqttConnectionHandler {

    private final DeviceConnectStatusListener deviceConnectStatusListener;
    private final DeviceMessageListener deviceMessageListener;
    private final ServerConnectStatusListener serverConnectStatusListener;
    private final ServerMessageListener serverMessageListener;

    /**
     * 处理连接
     */
    boolean connectHandler(MqttEndpoint endpoint) {
        final String clientId = endpoint.clientIdentifier();

        log.info("MQTT Server ====> connection request [clientId={}, remote={}]", clientId, endpoint.remoteAddress());

        if (!validateAuth(endpoint, clientId)) {
            return false;
        }

        // 按客户端类型（设备or服务端）分别处理
        if (clientId.startsWith(ServerClient.PREFIX)) {
            return serverConnectStatusListener.online(endpoint);
        } else {
            return deviceConnectStatusListener.online(endpoint);
        }
    }

    /**
     * Publish
     */
    void publishHandler(MqttEndpoint endpoint) {
        // QoS级别是0时，服务端不需要给客户端响应
        // QoS级别是1时，服务端需要通过 publishAcknowledge 方法回应一个 PUBACK 消息
        // QoS级别是2时，服务端需要通过 publishReceived 方法回应一个 PUBREC 消息
        endpoint.publishHandler(message -> {
            String clientId = endpoint.clientIdentifier();

            if (message.qosLevel() == MqttQoS.AT_LEAST_ONCE) {
                endpoint.publishAcknowledge(message.messageId());
            } else if (message.qosLevel() == MqttQoS.EXACTLY_ONCE) {
                endpoint.publishReceived(message.messageId());
            }

            if (clientId.startsWith(ServerClient.PREFIX)) {
                serverMessageListener.onMessage(endpoint, message);
            } else {
                deviceMessageListener.onMessage(endpoint, message);
            }

        }).publishReleaseHandler(messageId -> {
            log.info("MQTT Server ====> endpoint.publishComplete(messageId): {}", messageId);
            // 为了关闭QoS级别2传递，可以使用 publishComplete 方法用来发送 PUBCOMP 消息到客户端
            endpoint.publishComplete(messageId);
        });
    }

    /**
     * 初步认证验证
     */
    private boolean validateAuth(MqttEndpoint endpoint, String clientId) {
        MqttAuth auth = endpoint.auth();
        if (auth == null || StringUtils.isAnyBlank(auth.getUsername(), auth.getPassword())) {
            log.info("MQTT Server ====> Client connection failed [clientId={}], incorrect username or password", clientId);
            endpoint.reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD);
            return false;
        }

        return true;
    }

    /**
     * 订阅
     */
    void subscribeHandler(MqttEndpoint endpoint) {
        endpoint.subscribeHandler(subscribe -> {
            List<MqttQoS> grantedQosLevels = new ArrayList<>();

            for (MqttTopicSubscription s : subscribe.topicSubscriptions()) {
                log.info("MQTT Server ====> Client [{}] Subscription for {} with QoS {}", endpoint.clientIdentifier(), s.topicName(), s.qualityOfService());
                grantedQosLevels.add(s.qualityOfService());
            }

            // ack the subscriptions request
            endpoint.subscribeAcknowledge(subscribe.messageId(), grantedQosLevels);
        });
    }

    /**
     * 取消订阅
     */
    void unsubscribeHandler(MqttEndpoint endpoint) {
        endpoint.unsubscribeHandler(unsubscribe -> {
            for (String t : unsubscribe.topics()) {
                log.info("MQTT Server ====> Unsubscription for {}", t);
            }

            // ack the subscriptions request
            endpoint.unsubscribeAcknowledge(unsubscribe.messageId());
        });
    }

    /**
     * 断开连接
     */
    void closeHandler(MqttEndpoint endpoint) {
        endpoint.closeHandler(v -> {
            String clientId = endpoint.clientIdentifier();

            if (clientId.startsWith(ServerClient.PREFIX)) {
                serverConnectStatusListener.offline(endpoint);
                log.info("MQTT Server ====> Server [{}] Connection closed", clientId);
            } else {
                deviceConnectStatusListener.offline(endpoint);
                log.info("MQTT Server ====> Client [{}] Connection closed, Current device connection count [{}]",
                    clientId, MqttSessionManager.getInstance().getSessionCount());
            }
        });
    }

}
